package mq;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

public class BrokerServer {

    private static final int SERVER_PORT = 8080;
    private Socket socket;


    public BrokerServer(Socket socket) {
             this.socket = socket;
    }


    public void run(){
        System.out.println("启动存储节点");
        try{
            BufferedReader in=new BufferedReader(new InputStreamReader(
                    socket.getInputStream()));
            PrintWriter out=new PrintWriter(socket.getOutputStream());
            {
                while(true){
                  System.out.println("读数据");
                  String str =in.readLine();
                  if(str==null){
                      System.out.println("数据为空,就继续轮询读数据");
                      Thread.sleep(1000);
                      continue;
                  }
                  System.out.println("收到数据："+str);


                  //判断是生产者还是消费者
                    if(str.equals("consumer")){//CONSUME 表示要消费一条消息
                        //从消息队列中消费一条消息
                        String consume = Broker.consume();
                        out.println(consume);
                        out.flush();
                    }else if(str.contains("SEND:")){
                        //接受到的请求包含SEND；字符串 表示生产消息放到消息队列中
                        Broker.produce(str);
                    }else{
                        System.out.println("原始数据："+str+"没有遵循协议,不提供相关服务");
                        System.out.println("结束服务");
                        return;
                    }
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
        while(true){
            Socket socket = serverSocket.accept();
            new Thread((Runnable) new BrokerServer(socket)).start();
        }
    }
}


